package org.osomit.sacct.remoting.socket;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.Date;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.osomit.sacct.remoting.Listener;
import com.google.inject.Provider;
import com.google.inject.Inject;

/**
 * 
 * @author Jian Fang (John.Jian.Fang@gmail.com)
 *
 * Date: Jun 26, 2009
 */
public class SocketListener implements Listener {
    private int poolSize;
    private ExecutorService executor;
    private ServerSocket serverSocket;
    private int port;

    private volatile Boolean initlock = new Boolean(false);
    private boolean initialized = false;
    private volatile boolean paused = false;
    
    private volatile boolean running = true;
    private final static int MAX_INIT_WAIT = 20 * 1000;
 	private static int SHUTDOWN_WAIT_SECONDS = 20;

    private Provider<SocketWorker> socketWorkerProvider;

	private Log log = LogFactory.getLog(SocketListener.class);

    public Provider<SocketWorker> getSocketWorkerProvider() {
        return socketWorkerProvider;
    }

    @Inject
    public void setSocketWorkerProvider(Provider<SocketWorker> socketWorkerProvider) {
        this.socketWorkerProvider = socketWorkerProvider;
    }

    public int getPoolSize() {
        return poolSize;
    }

    public void setPoolSize(int poolSize) {
        this.poolSize = poolSize;
    }

    public int getPort() {
        return port;
    }

    public void setPort(int port) {
        this.port = port;
    }

    public Boolean isInitlock() {
        return initlock;
    }

    public void setInitlock(Boolean initlock) {
        this.initlock = initlock;
    }

    public boolean isInitialized() {
        return initialized;
    }

    public void setInitialized(boolean initialized) {
        this.initialized = initialized;
    }

    public boolean isRunning() {
        return running;
    }

    public void setRunning(boolean running) {
        this.running = running;
    }

    public boolean isPaused() {
        return paused;
    }

    public void setPaused(boolean paused) {
        this.paused = paused;
    }

    public SocketListener() {

    }

    public SocketListener(int port, int poolSize) {
        this.port = port;
        this.poolSize = poolSize;
		this.executor = Executors.newFixedThreadPool(this.poolSize);
    }

/*
    public void awaitInitialization()
    {
        synchronized (initlock) {
            long remainingWaitTime = MAX_INIT_WAIT;
            while (!initialized && remainingWaitTime > 0) {
                long start = System.currentTimeMillis();
                try {
                    initlock.wait(remainingWaitTime);
                } catch (InterruptedException e) {

                }
                remainingWaitTime -= Math.max(1, System.currentTimeMillis() - start);
            }

            if (!initialized){
                log.error("Timed out waiting for socket listener to initialize");
                throw new RuntimeException("Timed out waiting for socket listener to initialize");
            }
        }
    }
*/

    public void run() {
        Thread.currentThread().setName("SAcct Server Listener @ :" + port);
        Thread.currentThread().setPriority(Thread.NORM_PRIORITY+2);
        running = true;
        //urgly, should make the constructor injection with parameters working
        if(this.executor == null)
            this.executor = Executors.newFixedThreadPool(this.poolSize);
        try {
            serverSocket = new ServerSocket(port);
            serverSocket.setSoTimeout(1 * 1000);
        } catch (IOException e) {
            e.printStackTrace();
            // allow the thread to exit
        }

/*
        synchronized (initlock)
        {
            initialized = true;
            initlock.notify();
        }
*/

        log.debug("socket listener listening for requests @ :" + port);
        
        while (running) {
            try {
                Socket clientSocket = serverSocket.accept();
                log.debug("socket listener @ :" + port + " received connection at " + new Date());
                if (clientSocket != null) {
                    log.info("socket listener @ :" + port + " launching request processing thread");

                    SocketWorker worker = socketWorkerProvider.get();
                    worker.initalize(clientSocket);
                    this.executor.execute(worker);
                }
                else
                {
                    System.out.println("WARNING: Client socket was null!");
                }
            }
            catch (SocketTimeoutException se)
            {
                //do nothing and continue
//                log.warn("socket listener timed out");
            }
            catch (IOException e) {
                log.error("socket listener exception" +  e.getMessage());
            }
        }

        log.info("socket listener stopped listening @ :" + port);
    }


    public void start() {
        run();
    }

    public void stop() {
        running = false;
        log.info("Socket Listener is shutting down... ");

        executor.shutdown();
        try {
            executor.awaitTermination(SHUTDOWN_WAIT_SECONDS, TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            log.error("Socket Listener shutdown exception " + e.getLocalizedMessage());
            e.printStackTrace();
        }
    }

    public void pause() {
        paused = true;
    }

    public void resume() {
        paused = false;
    }

}
